package main

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/mjun1833/buildnotification/clients/dbclient/source"
	"gitee.com/mjun1833/buildnotification/clients/es"
	"gitee.com/mjun1833/buildnotification/clients/pushclient"
	"gitee.com/mjun1833/buildnotification/config"
	"gitee.com/mjun1833/buildnotification/entity"
	_ "gitee.com/mjun1833/buildnotification/keepalive"
	"gitee.com/mjun1833/buildnotification/model"
	"gitee.com/mjun1833/buildnotification/utils/dgfile/filehandler"
	"gitee.com/mjun1833/buildnotification/utils/dgfile/loggo"
	"gitee.com/mjun1833/buildnotification/utils/notification"
	"github.com/olivere/elastic/v7"
	"github.com/robfig/cron"
	log "github.com/sirupsen/logrus"
	"io/ioutil"
	"math"
	"math/big"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

var AddDayNum = "add_day_num"

const (
	OneDayTimestamp = int64(86400 * 1000)
)

type CaptureData struct {
	DataTypes []model.DataType
	Data      *model.CreateSubscribeNotificationArrReq
}

type LogData struct {
	Msg string `json:"msg"`
}

var (
	Apes                               = make([]*entity.APE, 0)
	Tollgates                          = make([]*entity.Tollgate, 0)
	Subscribes                         = make([]*entity.Subscribe, 0)
	DataChan                           = make(chan CaptureData, 1024)
	LocalView                          = &entity.ViewDatabase{}
	SendSuccessNotificationDataTypeMap = make(map[model.DataType]int)
	SendFailNotificationDataTypeMap    = make(map[model.DataType]int)
	SendNotificationMap                = make(map[SendStatus]int)
	Lock                               sync.RWMutex
	CronRule                           = "0/5 * * * * ?"
	AddTimestamp                       = int64(0)
	LimitQPSChan                       = make(chan struct{}, 1000000)
)

func main() {
	//readStatistics()
	//WriteStatistics()
	//limitQPS()
	//BuildData()
	//readNum()
	//writeNum()
	//go EsSendMessage()
	go selectData()
	//go buildTestData()
	waitStop()
}

var tempTime = time.Duration(200)
var TempChan = make(chan MyTest, 1024)
var sendArr = []string{"0"}
var tempNum = 0
var SendNum = 0

type MyTest struct {
	Id   string
	Name string
}

func EsSendMessage() {
	for {
		select {
		case temp := <-TempChan:
			errMap, err := sendEs(temp)
			if err == nil && len(errMap) == 0 {
				sendArr = append(sendArr, temp.Id)
				log.Infof("发送成功: %v", temp.Id)
			} else {
				if err != nil {
					log.Errorf("发送失败: %v,err: %v", temp.Id, err)
				} else if len(errMap) != 0 {
					log.Errorf("发送失败: %v,err: %v", temp.Id, errMap)
				}
			}
			num, _ := strconv.Atoi(temp.Id)
			SendNum = num
			writeNum()
		}
	}
}

func buildTestData() {
	for {
		select {
		case <-time.After(time.Millisecond * tempTime):
			TempChan <- MyTest{
				Id:   strconv.Itoa(tempNum),
				Name: "name" + strconv.Itoa(tempNum),
			}
			tempNum++
		}
	}
}

func selectData() {
	for {
		select {
		case <-time.After(time.Millisecond * tempTime):
			//go func() {
			//	id := sendArr[RangeRand(0, int64(len(sendArr)-1))]
			//	if id == "0" {
			//		return
			//	}
			//	res, err := es.GetInstance().Search().Index("test").IgnoreUnavailable(true).Query(elastic.NewIdsQuery().Ids(id)).Do(context.Background())
			//	if err != nil || res.Hits.TotalHits.Value == 0 {
			//		log.Errorf("查询失败: %v,err: %v", id, err)
			//		return
			//	}
			//	log.Infof("查询成功: %v", id)
			//}()

			go func() {
				res, err := es.GetInstance().CatHealth().Do(context.Background())
				if err != nil || len(res) == 0 {
					log.Errorf("查询 health 失败,err: %v", err)
					return
				}

				for _, re := range res {
					log.Infof("查询 health 成功,status: %v,time: %v", re.Status, re.Timestamp)
				}
			}()
		}
	}
}

func sendEs(data MyTest) (map[string]error, error) {
	bulkRequest := es.GetInstance().Bulk()
	bulkRequest = bulkRequest.Add(elastic.NewBulkIndexRequest().Id(data.Id).Index("test").Doc(&data))

	res, err := bulkRequest.Do(context.TODO())
	if err != nil {
		return nil, err
	}

	if len(res.Failed()) == 0 {
		return nil, err
	}

	errMap := map[string]error{}
	for i := 0; i < len(res.Failed()); i++ {
		errMap[res.Failed()[i].Id] = errors.New(res.Failed()[i].Error.Reason)
	}

	return errMap, err
}

func waitStop() {
	sc := make(chan os.Signal, 1)
	signal.Notify(sc,
		syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)

	sig := <-sc
	log.Printf("exit: signal=<%d>.\n", sig)
	switch sig {
	case syscall.SIGTERM:
		//err := writeStatistics()
		//if err != nil {
		//	log.Errorf(err.Error())
		//}
		writeNum()
		log.Println("exit: bye :-).")
		os.Exit(0)
	default:
		//err := writeStatistics()
		//if err != nil {
		//	log.Errorf(err.Error())
		//}
		writeNum()
		log.Println("exit: bye :-(.")
		os.Exit(1)
	}
}

func BuildData() {
	engine := source.GetInstance().GetDBEngine()

	//apes := make([]*entity.APE, 0)
	err := engine.Table(entity.TableNameAPE).Find(&Apes)
	if err != nil {
		panic(err)
	}
	for _, ape := range Apes {
		entity.ApeMap[ape.ApeID] = ape
	}

	//tollgates := make([]*entity.Tollgate, 0)
	err = engine.Table(entity.TableNameTollgate).Find(&Tollgates)
	if err != nil {
		panic(err)
	}
	for _, tollgate := range Tollgates {
		entity.TollgateMap[tollgate.TollgateID] = tollgate
	}

	//subscribes := make([]*entity.Subscribe, 0)

	exist, err := engine.Table(entity.TableNameViewDatabase).Where("view_database_code = ?", config.GetInstance().GetString("buildConf.localViewCode")).Get(LocalView)
	if err != nil {
		panic(err)
	}
	if !exist {
		panic(fmt.Errorf("code: %v不存在", config.GetInstance().GetString("buildConf.localViewCode")))
	}

	err = engine.Table(entity.TableNameSubscribe).Where("target_view_id = ?", LocalView.ViewDatabaseId).Find(&Subscribes)
	if err != nil {
		panic(err)
	}

	kv := entity.KVEntity{}
	kvExist, err := engine.Table("kv").Where("key = ?", AddDayNum).Get(&kv)
	if err != nil {
		panic(err)
	}
	if !kvExist {
		kv = entity.KVEntity{
			Key:   AddDayNum,
			Value: "0",
		}
		_, err = engine.Table("kv").Insert(&kv)
		if err != nil {
			panic(err)
		}
	}
	dayNum, err := strconv.Atoi(kv.Value)
	if err != nil {
		panic(err)
	}
	fmt.Println("-------------")
	fmt.Printf("day: %v\n", dayNum)
	fmt.Println("-------------")

	AddTimestamp = OneDayTimestamp * int64(dayNum)

	go func() {
		err := GetData(DataChan)
		if err != nil {
			panic(err)
		}
	}()

	go func() {
		SendData(DataChan)
	}()
}

// 读取人车非脸的数据
func GetData(dataChan chan CaptureData) error {
	// 文件读取
	loggo.InitDefaultLog() // 初始化日志组件
	handler := &filehandler.FileScanner{
		BatchNum:          config.GetInstance().GetInt("buildConf.readFileNum"), // 同时处理的文件个数
		BaseDir:           config.GetInstance().GetString("buildConf.dir"),      // 扫描文件的根目录
		FileType:          "log",                                                // 文件类型
		RemoveHandledFile: false,                                                // 是否删除已处理的文件
	}
	handler.FileValidation = FileValidation
	handler.HandleByteFunc = func(line []byte) error {
		return HandleByteFunc(line, dataChan)
	}
	handler.Init()                           // 初始化扫描器
	return handler.LoopScan(3 * time.Second) // 开始扫描
}

func FileValidation(info os.FileInfo) bool {
	return true
}

func HandleByteFunc(line []byte, dataChan chan CaptureData) error {
	line1 := strings.ReplaceAll(string(line), " ", "")
	line2 := strings.ReplaceAll(line1, "\r", "")
	line3 := strings.ReplaceAll(line2, "\n", "")
	//line4 := strings.ReplaceAll(line3, "\\", "")

	if line3 == "" || line3 == "\"" {
		return nil
	}

	//logData := &LogData{}
	//err := json.Unmarshal([]byte(line3), logData)
	//if err != nil {
	//	log.Errorf("line: %v,unmarshal fail,err: %v", string(line), err)
	//	return nil
	//}
	//
	//split := strings.Split(logData.Msg, "|")
	//if len(split) < 5 {
	//	log.Errorf("line: %v,split fail,err: %v", string(line), err)
	//	return nil
	//}
	//
	//reqStrs := strings.SplitN(split[4], ":", 2)
	//if len(reqStrs) < 2 {
	//	log.Errorf("line: %v,reqStr split fail,err: %v", string(line), err)
	//	return nil
	//}

	notifications := &model.CreateSubscribeNotificationArrReq{}
	err := json.Unmarshal([]byte(line3), &notifications)
	if err != nil {
		log.Errorf("line: %v,unmarshal fail,err: %v", string(line), err)
		return nil
	}

	if len(notifications.SubscribeNotificationListObject.SubscribeNotificationObject) == 0 {
		log.Warningf("通知为空: %v", string(line))
		return nil
	}

	dataChan <- CaptureData{
		Data: notifications,
	}

	return nil
}

// 组装和发送通知
func SendData(dataChan chan CaptureData) {
	threadNum := make(chan struct{}, config.GetInstance().GetInt("buildConf.maxSendThread"))
	for {
		select {
		case captureData := <-dataChan:
			for _, subscribe := range Subscribes {
				if subscribe.TargetViewID == LocalView.ViewDatabaseId {
					notifications, dataTypes, err := notification.BuildSubscribeNotificationArrReq2(subscribe, captureData.Data, RandomApeID(), RandomTollgateID(), config.GetInstance().GetBool("buildConf.isBuild"), config.GetInstance().GetBool("buildConf.isBuildTime"), config.GetInstance().GetBool("buildConf.isBuildApeExtend"), AddTimestamp)
					if err != nil {
						log.Errorf("构建通知失败: %v", err)
						continue
					}

					if notifications == nil {
						marshal, _ := json.Marshal(captureData.Data)
						log.Warningf("过滤数据：%v,AddTimestamp: %v", string(marshal), AddTimestamp)
						continue
					}

					// 判断这个订阅数据是否需要转发
					flag := false
					for _, dataType := range dataTypes {
						if strings.Contains(subscribe.SubscribeDetail, string(dataType)) {
							flag = true
							break
						}
					}

					if !flag {
						continue
					}

					// 是否只构造face
					if config.GetInstance().GetBool("buildConf.isBuildSingleFace") {
						if len(dataTypes) != 1 || dataTypes[0] != model.DataTypeFace {
							continue
						}
					}

					threadNum <- struct{}{}
					<-LimitQPSChan
					go func(temp interface{}, dataTypes []model.DataType, addr string) {
						start := time.Now()
						defer func() {
							log.Infof("发送通知花费了: %v秒", time.Since(start).Seconds())
							<-threadNum
						}()

						err = pushclient.GetInstance().PostSend(addr, config.GetInstance().GetString("buildConf.localViewCode"), temp)
						if err != nil {
							log.Errorf("发送通知失败: %v", err)
							CalculateStatistics(dataTypes, SendStatusFail)
							return
						}

						CalculateStatistics(dataTypes, SendStatusSuccess)

					}(notifications, dataTypes, subscribe.ReceiveAddr)
				}
			}
		}
	}
}

func RandomApeID() string {
	return Apes[RangeRand(0, int64(len(Apes)-1))].ApeID
}

func RandomTollgateID() string {
	return Tollgates[RangeRand(0, int64(len(Tollgates)-1))].TollgateID
}

func RangeRand(min, max int64) int64 {
	if min > max {
		return 0
	}

	if min < 0 {
		f64Min := math.Abs(float64(min))
		i64Min := int64(f64Min)
		result, _ := rand.Int(rand.Reader, big.NewInt(max+1+i64Min))

		return result.Int64() - i64Min
	} else {
		result, _ := rand.Int(rand.Reader, big.NewInt(max-min+1))
		return min + result.Int64()
	}
}

type SendStatus string

const (
	SendStatusSuccess SendStatus = "success"
	SendStatusFail    SendStatus = "fail"
)

func CalculateStatistics(dataTypes []model.DataType, sendStatus SendStatus) {
	Lock.Lock()
	defer Lock.Unlock()
	switch sendStatus {
	case SendStatusSuccess:
		for _, dataType := range dataTypes {
			SendSuccessNotificationDataTypeMap[dataType] = 1 + SendSuccessNotificationDataTypeMap[dataType]
		}
		SendNotificationMap[SendStatusSuccess] = 1 + SendNotificationMap[SendStatusSuccess]
	case SendStatusFail:
		for _, dataType := range dataTypes {
			SendFailNotificationDataTypeMap[dataType] = 1 + SendFailNotificationDataTypeMap[dataType]
		}
		SendNotificationMap[SendStatusFail] = 1 + SendNotificationMap[SendStatusFail]
	}
}

func WriteStatistics() {
	c := cron.New()
	c.AddFunc(CronRule, func() {
		err := writeStatistics()
		if err != nil {
			log.Errorf(err.Error())
		}
	})
	c.AddFunc(config.GetInstance().GetString("buildConf.changAddDayNumCronRule"), func() {
		err := ChangeAddTimestamp()
		if err != nil {
			log.Errorf(err.Error())
		}
	})
	c.Start()
}

func writeStatistics() error {
	Lock.Lock()
	defer Lock.Unlock()

	log.Infof("开始写入统计信息")

	failMarshal, err := json.Marshal(SendFailNotificationDataTypeMap)
	if err != nil {
		return err
	}

	successMarshal, err := json.Marshal(SendSuccessNotificationDataTypeMap)
	if err != nil {
		return err
	}

	notificationMarshal, err := json.Marshal(SendNotificationMap)
	if err != nil {
		return err
	}

	err = ioutil.WriteFile(config.GetInstance().GetString("buildConf.dir")+"//SendFailNotification.log", failMarshal, 0666)
	if err != nil {
		return err
	}

	err = ioutil.WriteFile(config.GetInstance().GetString("buildConf.dir")+"//SendSuccessNotification.log", successMarshal, 0666)
	if err != nil {
		return err
	}

	err = ioutil.WriteFile(config.GetInstance().GetString("buildConf.dir")+"//SendNotification.log", notificationMarshal, 0666)
	if err != nil {
		return err
	}

	log.Infof("开始完成统计信息写入")

	return nil
}

func readStatistics() {
	Lock.Lock()
	defer Lock.Unlock()

	go func() {
		failBytes, err := ioutil.ReadFile(config.GetInstance().GetString("buildConf.dir") + "//SendFailNotification.log")
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		err = json.Unmarshal(failBytes, &SendFailNotificationDataTypeMap)
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		log.Infof("本地fail统计读取成功")
	}()

	go func() {
		successBytes, err := ioutil.ReadFile(config.GetInstance().GetString("buildConf.dir") + "//SendSuccessNotification.log")
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		err = json.Unmarshal(successBytes, &SendSuccessNotificationDataTypeMap)
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		log.Infof("本地success统计读取成功")
	}()

	go func() {
		notificationBytes, err := ioutil.ReadFile(config.GetInstance().GetString("buildConf.dir") + "//SendNotification.log")
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		err = json.Unmarshal(notificationBytes, &SendNotificationMap)
		if err != nil {
			log.Errorf(err.Error())
			return
		}

		log.Infof("本地notification统计读取成功")
	}()
}

func limitQPS() {
	limit := config.GetInstance().GetInt("buildConf.limitQPS")
	go func() {
		for {
			select {
			case <-time.After(time.Second * 1):
				for i := 0; i < limit; i++ {
					LimitQPSChan <- struct{}{}
				}
			}
		}
	}()
}

func ChangeAddTimestamp() error {
	engine := source.GetInstance().GetDBEngine()

	kv := entity.KVEntity{}
	kvExist, err := engine.Table("kv").Where("key = ?", AddDayNum).Get(&kv)
	if err != nil {
		return err
	}
	if !kvExist {
		kv = entity.KVEntity{
			Key:   AddDayNum,
			Value: "0",
		}
		_, err = engine.Table("kv").Insert(&kv)
		if err != nil {
			panic(err)
		}
	}

	dayNum, err := strconv.Atoi(kv.Value)
	if err != nil {
		panic(err)
	}

	newDayNum := dayNum + 1
	tempKv := entity.KVEntity{
		Key:   AddDayNum,
		Value: strconv.Itoa(newDayNum),
	}
	_, err = engine.Table("kv").Where("key = ?", tempKv.Key).Update(&tempKv)
	if err != nil {
		panic(err)
	}

	AddTimestamp = OneDayTimestamp * int64(newDayNum)

	return nil
}

func writeNum() error {
	Lock.Lock()
	defer Lock.Unlock()

	log.Infof("开始写入num信息")

	err := ioutil.WriteFile(config.GetInstance().GetString("buildConf.dir")+"//num.log", []byte(strconv.Itoa(SendNum)), 0666)
	if err != nil {
		return err
	}

	log.Infof("完成num写入")

	return nil
}

func readNum() {
	Lock.Lock()
	defer Lock.Unlock()

	failBytes, err := ioutil.ReadFile(config.GetInstance().GetString("buildConf.dir") + "//num.log")
	if err != nil {
		log.Errorf(err.Error())
		return
	}

	num, err := strconv.Atoi(string(failBytes))
	if err != nil {
		log.Errorf(err.Error())
	}
	tempNum = num

	log.Infof("num读取成功")
}
